package demo1;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

/**
 * @Author: wzg
 * @Date: 2021/4/14 下午4:50
 */
public class Consumer {
    public static void main(String[] args) throws InterruptedException {
        Map<String,Object> configs = new HashMap<>();
        configs.put("bootstrap.servers", "192.168.50.117:9092");
        configs.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configs.put("group.id", "consumer.demo");
        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(configs);
        Pattern pattern = Pattern.compile("topic_[0-9]");

        List<String> topics = Arrays.asList("topic_1");
        consumer.subscribe(topics, new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                collection.forEach(p->{
                    System.out.println("剥夺的分区："+p.partition());
                });
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                collection.forEach(p->{
                    System.out.println(p.partition());
                });
            }
        });
        for (int i = 0; i < 100; i++) {
            final ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(3));

            final Iterable<ConsumerRecord<Integer, String>> topic1Iterable = records.records("topic_1");

            topic1Iterable.forEach(record -> {
                System.out.println("========================================");
                System.out.println("消息头字段：" + Arrays.toString(record.headers().toArray()));
                System.out.println("消息的key：" + record.key());
                System.out.println("消息的偏移量：" + record.offset());
                System.out.println("消息的分区号：" + record.partition());
                System.out.println("消息的序列化key字节数：" + record.serializedKeySize());
                System.out.println("消息的序列化value字节数：" + record.serializedValueSize());
                System.out.println("消息的时间戳：" + record.timestamp());
                System.out.println("消息的时间戳类型：" + record.timestampType());
                System.out.println("消息的主题：" + record.topic());
                System.out.println("消息的值：" + record.value());
            });
            TimeUnit.SECONDS.sleep(1);
        }
        consumer.close();
    }
}
